/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.just.dusin.ooii.component.standard;
import java.util.List;
import java.util.logging.Level;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import java.util.Properties;
import org.apache.log4j.Logger;
import org.just.dusin.ooii.component.Sink;
import org.dusin.webcoal.exception.OOIIException;

/**
 *
 * @author Administrator
 */
public class AMQSink implements Sink{
    
    private String brokerName="broke";
    private String brokerURL = "tcp://localhost:61616";
    private String queueName="coal";
    
    
    
    private String user;
    private String password;
    public String getBrokerName() {
        return brokerName;
    }

    public void setBrokerName(String brokerName) {
        this.brokerName = brokerName;
    }

    public String getBrokerURL() {
        return brokerURL;
    }

    public void setBrokerURL(String brokerURL) {
        this.brokerURL = brokerURL;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }
    
    
    
    private static Logger logger = Logger.getLogger(AMQSink.class.getName());
    

    @Override
    public void initComponent(Properties props) {
        
    }

    @Override
    public void process(String msg) throws OOIIException {
        Connection connection=null;
        Session session=null;
        MessageProducer producer = null;
        try{
            
            ConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);    	
            connection = factory.createConnection(user,password);
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            producer = session.createProducer(null);
            Destination destination = session.createQueue(queueName );
            Message message = session.createTextMessage(msg);   
            Destination respDest = session.createQueue("response");
            message.setJMSReplyTo(respDest);
            producer.send(destination, message);
            logger.info("处理消息"+msg);
        }catch(JMSException e){
            logger.warn("抛出异常"+e.getMessage());
            throw new OOIIException(e);
            
        }finally{
            if(producer!=null){
                try {
                    producer.close();
                } catch (JMSException ex) {
                    logger.error("producer 关闭错误");
                }
            }
            if(session!=null){
                try {
                    session.close();
                } catch (JMSException ex) {
                    logger.error("session 关闭错误");
                }
            }
             if(connection != null) {
                try {
                    connection.close();
                } catch (JMSException ex) {
                    logger.error("connection 关闭错误");
                }
            }
        }
    }

    @Override
    public void process(List<String> msgs) throws OOIIException{
    }

    @Override
    public void destroyComponent() throws Exception {
       
    }
    
}
